package com.storm.demo;

import org.apache.storm.Config;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.topology.TopologyBuilder;

/**
 * =================================
 * <p>
 * Created by cjj on 2018/10/23.
 * <p>
 * 描述: 组织各个处理组件形成一个完整的处理流程，就是所谓的topology（类似mapreduce的job）区别在于job执行完就停止了）
 * 并且将topolgy提交给storm运行，topolgy提交到集群会永不休止的取运行，除非人为去停止
 */

public class StormDemoMain {


    public static void main(String[] args) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {

        TopologyBuilder topologyBuilder = new TopologyBuilder();

        //将spout设置到topology中 并发度 4
        topologyBuilder.setSpout("randomSpout", new RandomWordSpout(), 4);

        //将小写转大写组件bolt添加到topology，并指定它接受randomSpout的消息 并发度 4
        //shuffleGrouping 随机分组（把spout分给blot）
        topologyBuilder.setBolt("upperBlot", new UpperBlot(), 4).shuffleGrouping("randomSpout");

        //后缀组件blot添加到topology
        topologyBuilder.setBolt("suffixBlot", new SuffixBlot(), 4).shuffleGrouping("randomSpout");

        StormTopology topology = topologyBuilder.createTopology();

        //配置topology在集群中的运行参数
        Config config = new Config();
        config.setNumWorkers(4);
        config.setDebug(true);
        config.setNumAckers(0);

        StormSubmitter.submitTopology("", config, topology);


    }

}
